package com.song.client;

import com.song.NacosServiceRegistry.LoadBalancer;
import com.song.NacosServiceRegistry.NacosNacosServiceRegistryImpl;
import com.song.NacosServiceRegistry.NacosServiceRegistry;
import com.song.Utils.SingletonFactory;
import com.song.enumeration.RpcError;
import com.song.exception.RpcException;
import com.song.transport.*;
import io.netty.channel.*;
import lombok.Data;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

/**
 * 责任链模式
 * 每个处理器都会对数据进行加工，并将处理后的数据传给下一个处理器。
 * 代码中的 CommonEncoder、CommonDecoder和NettyServerHandler 分别就是编码器，解码器和数据处理器
 */
@Slf4j(topic = "c.NettyClient")
@Data
@Setter
public class NettyClient implements RpcClient {

    private String host;
    private int port;
    private CommonSerializer serializer;

    private LoadBalancer loadBalancer;
    NacosServiceRegistry nacosServiceRegistry;

    private final UnprocessedRequests unprocessedRequests;//常量 全局唯一  final 声明时未初始化  那么必须保证所有构造器都有初始化

    public NettyClient() {
        this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
    }

    public NettyClient(LoadBalancer loadBalancer) {
        this.nacosServiceRegistry = new NacosNacosServiceRegistryImpl(loadBalancer);
        this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
    }

    public NettyClient(String host, int port) {
        this.host = host;
        this.port = port;
        this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);;
    }

    @Override
    public CompletableFuture<RPCResponse> sendRequest(RPCRequest rpcRequest) {
        if(serializer == null) {
            log.error("未设置序列化器");
            throw new RpcException(RpcError.SERIALIZER_NOT_FOUND);
        }
//        AtomicReference<Object> result = new AtomicReference<>();
        CompletableFuture<RPCResponse> res = new CompletableFuture<>();//异步等待获取计算结果
        try {
            InetSocketAddress inetSocketAddress = nacosServiceRegistry.lookupService(rpcRequest.getInterfaceName(), rpcRequest);

            //*****
            //Netty获取channel 此类的connect方法中实现了CompletableFuture的complete方法获取监听的channel返回得到的Response
            //此处channel获取并传出RPCResponse
            Channel channel = ChannelProvider.get(inetSocketAddress, serializer);

            unprocessedRequests.put(rpcRequest.getRequestId(), res);//使用了ConcurrentHashMap结构 来存放Response消息

            channel.writeAndFlush(rpcRequest).addListener((ChannelFutureListener)future1 -> {//通过Netty通道发送数据并且 开启通道异步监听
                if(future1.isSuccess()) {
                    log.info(String.format("客户端发送消息：%s", rpcRequest.toString()));
                } else {
                    future1.channel().close();
                    res.completeExceptionally(future1.cause());//CompletableFuture结果异常处理(Throwable)
                    log.error("发送消息时发送错误：", future1.cause());
                }
            });

//            AttributeKey<RPCResponse> key = AttributeKey.valueOf("rpcResponse");//通过 AttributeKey 的方式  阻塞获得  返回结果
//            RPCResponse rpcResponse = channel.attr(key).get();//处理收到的RPCResponse
//            return rpcResponse.getData();//返回RPCResponse的具体信息
        } catch (InterruptedException e) {
            unprocessedRequests.remove(rpcRequest.getRequestId());
            log.error("发送消息时有错误发生: ", e);
        }
        return res;
    }
}
